package com.klun.hdfs.hbase.driver;

import com.klun.hdfs.hbase.mapper.ReadFruitMapper;
import com.klun.hdfs.hbase.reducer.WriteFruitMRReducer;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;

public class Fruit2FruitMRRunner extends Configured implements Tool {

    //组装job
    @Override
    public int run(String[] strings) throws Exception {
        //
        Configuration conf = this.getConf();
        Job job = Job.getInstance(conf);
        job.setJarByClass(Fruit2FruitMRRunner.class);

        // 配置Job
        Scan scan = new Scan();
        scan.setCacheBlocks(false);
        scan.setCaching(500);

        // 设置mapper
        TableMapReduceUtil.initTableMapperJob(
                "fruit",
                scan,
                ReadFruitMapper.class,
                ImmutableBytesWritable.class,
                Put.class,
                job
        );

        // 设置 Reducer
        TableMapReduceUtil.initTableReducerJob("fruit_mr", WriteFruitMRReducer.class,job);
        // 设置 reducer 的数量，最少 1个
        job.setNumReduceTasks(1);
        boolean b = job.waitForCompletion(true);
        if(!b){
            throw new IOException("job running with error");
        }
        return b?0:1;
    }

    /**
     * 主运行函数
     * @param args
     */
    public static void main(String[] args) throws Exception {
        Configuration conf = HBaseConfiguration.create();
        int run = ToolRunner.run(conf, new Fruit2FruitMRRunner(), args);
        System.out.println(run);
        //maven-share-plugin

        Map<String,Long> map = new HashMap();
        Long a = new Long(0);
        map.put("a",a);

        System.out.println(map.get("a").compareTo(0L));
        System.out.println(map.get("a").compareTo(new Long(0l)));
        System.out.println(map.get("a").compareTo(new Long(0)));

        BigDecimal b = new BigDecimal(0);
        System.out.println(b.compareTo(BigDecimal.ZERO));
        System.out.println(b.equals(BigDecimal.ZERO));

    }

}
